package cn.gwm.flink.streaming.client;

import cn.gwm.utils.ConfigLoader;
import cn.gwm.utils.StringUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.jdbc.PhoenixDriver;

import java.io.IOException;
import java.util.*;

/**
 * @author Liu.Nanfang
 * @date 2022/8/15 14:08
 * @description
 */
public class HbaseClient implements AutoCloseable {
    private static Configuration conf = HBaseConfiguration.create();
    private static Admin admin;
    public static Connection conn;

    private static ThreadLocal<Connection> connThread = new ThreadLocal<Connection>();

    static {
//        Configuration conf = HBaseConfiguration.create();
//        conf.set("hbase.zookeeper.quorum", ConfigLoader.get("zookeeper.quorum", ""));
//        conf.set("hbase.zookeeper.property.clientPort", ConfigLoader.get("zookeeper.clientPort"));
        conf.set("hbase.zookeeper.quorum", "10.31.8.112,10.31.8.113,10.31.8.114");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.setInt("hbase.rpc.timeout", 3000);
        conf.setInt("hbase.client.operation.timeout", 3000);
        conf.setInt("hbase.client.scanner.timeout.period", 3000);
        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = getConn().getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static Connection getConn() {
        Connection conn = connThread.get();
        if (conn == null) {
            try {
                conn = ConnectionFactory.createConnection(conf);
                connThread.set(conn);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return conn;
    }

    public static void createTable(String tableName, String... columnFamilies) throws IOException {
        TableName tablename = TableName.valueOf(tableName);
        Admin admin = getConn().getAdmin();
        if (admin.tableExists(tablename)) {
            System.out.println("Table Exists");
        } else {
            System.out.println("Start create table");
            HTableDescriptor tableDescriptor = new HTableDescriptor(tablename);
            for (String columnFamliy : columnFamilies) {
                HTableDescriptor column = tableDescriptor.addFamily(new HColumnDescriptor(columnFamliy));
            }
            admin.createTable(tableDescriptor);
            System.out.println("Create Table success");
        }
    }

    /**
     * 获取一列获取一行数据
     *
     * @param tableName
     * @param rowKey
     * @param famliyName
     * @param column
     * @return
     * @throws IOException
     */
    public static String getData(String tableName, String rowKey, String famliyName, String column) throws IOException {
        Table table = getConn().getTable(TableName.valueOf(tableName));
        byte[] row = Bytes.toBytes(rowKey);
        Get get = new Get(row);
        Result result = table.get(get);
        byte[] resultValue = result.getValue(famliyName.getBytes(), column.getBytes());
        if (null == resultValue) {
            return null;
        }
        return new String(resultValue);
    }


    /**
     * 获取一行的所有数据并且排序
     *
     * @param tableName 表名
     * @param rowKey    列名
     * @throws IOException
     */
    public static List<Map.Entry> getRow(String tableName, String rowKey) throws IOException {
//        System.setProperty("hadoop.home.dir", "D:\\Program\\hadoop");
        Table table = getConn().getTable(TableName.valueOf(tableName));
        byte[] row = Bytes.toBytes(rowKey);
        Get get = new Get(row);
        Result r = table.get(get);
        HashMap<String, String> rst = new HashMap<>();
        if (r.isEmpty()) {
            return null;
        }
        for (Cell cell : r.listCells()) {
            String key = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            rst.put(key, value);
        }

        List<Map.Entry> ans = new ArrayList<>();
        ans.addAll(rst.entrySet());

        Collections.sort(ans, (m1, m2) -> new Double((Double) m1.getValue() - (Double) m2.getValue()).intValue());

        return ans;
    }

    /**
     * 向对应列添加数据
     *
     * @param tablename  表名
     * @param rowkey     行号
     * @param famliyname 列族名
     * @param column     列名
     * @param data       数据
     * @throws Exception
     */
    public static void putData(String tablename, String rowkey, String famliyname, String column, String data) throws Exception {
        Table table = getConn().getTable(TableName.valueOf(tablename));
        Put put = new Put(rowkey.getBytes());
        put.addColumn(famliyname.getBytes(), column.getBytes(), data.getBytes());
        table.put(put);
    }

    /**
     * 取出表中所有的key
     *
     * @param tableName
     * @return
     */
    public static List<String> getAllKey(String tableName) throws IOException {
        List<String> keys = new ArrayList<>();
        try {
            Scan scan = new Scan();
            Table table = HbaseClient.getConn().getTable(TableName.valueOf(tableName));
            ResultScanner scanner = table.getScanner(scan);
            for (Result r : scanner) {
                keys.add(new String(r.getRow()));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return keys;
    }

//     scan 'ods_gps_es13',{STARTROW=>'LGWEEUA50NC0027870009977880761',STOPROW=>'LGWEEUA50NC0027870000097880761'}

    public static List<Map.Entry> getRowData(String tableName, String startRowKey, String endRowKey) throws IOException {
        Table table = HbaseClient.getConn().getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRowKey));
        scan.setStopRow(Bytes.toBytes(endRowKey));
        ResultScanner results = table.getScanner(scan);
        Map<String, Result> map = new HashMap<>();
        HashMap<String, String> rst = new HashMap<>();
        List<Map.Entry> list = new ArrayList<>();
        for (Result result : results) {
            String key = new String(result.getRow());
            map.put(key, result);
            List<Cell> listCells = result.listCells();
            for (Cell cell : listCells) {
                String key1 = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                rst.put(key1, value);
                System.out.println("rst = " + rst);
            }
            list.addAll(rst.entrySet());
        }


//        list.addAll(map.entrySet());


        return list;
    }

    public static void main(String[] args) throws Exception {
        String sK = "LGWEEUA52NE202154" + StringUtil.reverse(String.valueOf("1670831818000"));
        String eK = "LGWEEUA52NE202154" + StringUtil.reverse(String.valueOf("1670831823000"));
        // LGWEEUA52NE2021540002281380761
//        List<Map.Entry> list = getRowData("ods_gps_es13", "LGWCESHI2022090710000011750761", "LGWCESHI2022090710009977880761");
//        System.out.println(list);
        String ad = HbaseClient.getData("DWS_RESERVE_CHARGE_AD_CODE", " LGWEEUA5XNE204000", "C1", "adCode");
        List a = HbaseClient.getRow("DWS_RESERVE_CHARGE_AD_CODE", "LGWEEUA5XNE204000");
        System.out.println(a);
        HbaseClient.putData("DWS_RESERVE_CHARGE_AD_CODE", "LGWEEUA5XNE204000", "C1", "adCode", "371103");
//        HbaseClient.putData("DWS_RESERVE_CHARGE_AD_CODE", "LGWEEUA5XNE204000", "C1", "adCode", "371102");
        List abb = HbaseClient.getRow("DWS_RESERVE_CHARGE_AD_CODE", "LGWEEUA5XNE204000");
        System.out.println(ad);
        System.out.println(abb);

        List<Map.Entry> cc = HbaseClient.getRow("DWS_RESERVE_CHARGE_AD_CODE", "LGWEEUA5XNE204000");
        if (cc.size() > 0) {
            cc.get(0).getValue();
        }


    }


    @Override
    public void close() throws Exception {
        Connection connection = connThread.get();
        if (connection != null) {
            connection.close();
        }
    }
}
